package org.lzfto.flink.demo.business.flinkTest.flinkObjects.function.agg;

import java.math.BigDecimal;

import org.apache.flink.api.common.functions.AggregateFunction;
import org.lzfto.flink.demo.business.orders.OrderCommon;
import org.lzfto.flink.demo.business.orders.SimpleOrderAgg;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.alibaba.fastjson2.JSON;

/**
*基于轻度汇总的二次汇总. 
*@author lzfto
*@创建时间 2025年6月20日 09:35:52
*/
public class AggOrderAggFunction implements AggregateFunction<SimpleOrderAgg, SimpleOrderAgg, SimpleOrderAgg> {
	private static final long serialVersionUID = 1L;
	
	private final Logger LOGGER = LoggerFactory.getLogger(getClass());
	private String grpType;
	
	
	public AggOrderAggFunction(String grpType) {
		this.grpType=grpType;
	}

	@Override
	public SimpleOrderAgg createAccumulator() {
		SimpleOrderAgg acc = new SimpleOrderAgg();
		acc.setCount(0);
		acc.setTotal(BigDecimal.ZERO);
		return acc;
	}

	@Override
	public SimpleOrderAgg add(SimpleOrderAgg value, SimpleOrderAgg acc) {		
		LOGGER.info("基于轻度汇总的汇总:{}   ,   {}",JSON.toJSONString(value),JSON.toJSONString(acc));
		if (grpType.equals(OrderCommon.GRP_TYPE_SEX)) {
			if (acc.getSex()==null) {
				acc.setSex(value.getSex());
			}
		}
		else if (grpType.equals(OrderCommon.GRP_TYPE_ZONE)) {
			if (acc.getZoneName()==null) {
				acc.setZoneName(value.getZoneName());
			}
		}
		else {
			if (acc.getZoneName()==null || acc.getSex()==null) {
				acc.setZoneName(value.getZoneName());
				acc.setSex(value.getSex());				
			}			
		}
		
		acc.setCount(acc.getCount() + value.getCount());
		acc.setTotal(acc.getTotal().add(value.getTotal()));
		//LOGGER.info("小汇总订单汇总结果:"+JSON.toJSONString(acc));
		return acc;
	}

	@Override
	public SimpleOrderAgg getResult(SimpleOrderAgg accumulator) {
		return accumulator;
	}

	@Override
	public SimpleOrderAgg merge(SimpleOrderAgg a, SimpleOrderAgg b) {
		LOGGER.info("合并操作:"+JSON.toJSONString(a)+" "+JSON.toJSONString(b));
		a.setCount(a.getCount() + b.getCount());
		a.setTotal(a.getTotal().add(b.getTotal()));
		return a;
	}
}
